package demo.db.elasticSearch.highClient;

import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.mapper.RangeFieldMapper.Range;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.profile.ProfileResult;
import org.elasticsearch.search.profile.ProfileShardResult;
import org.elasticsearch.search.profile.query.CollectorResult;
import org.elasticsearch.search.profile.query.QueryProfileShardResult;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.term.TermSuggestion;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class EsUtils {

    private static Logger logger = LoggerFactory.getLogger(EsUtils.class);

    public static int maxConnPerRoute = 10;

    public static int maxConnTotal = 30;

    public static ActionListener<IndexResponse> defaultIndexResponseActionListener = new ActionListener<IndexResponse>() {
        // Called when the execution is successfully completed.
        @Override
        public void onResponse(IndexResponse indexResponse) {
            logger.info("IndexResponse = {}", indexResponse);
        }

        // Called when the whole IndexRequest fails
        @Override
        public void onFailure(Exception e) {
            logger.error("", e);
        }
    };

    public static ActionListener<GetResponse> defaultGetResponseActionListener = new ActionListener<GetResponse>() {
        // Called when the execution is successfully completed.
        @Override
        public void onResponse(GetResponse getResponse) {
            logger.info("getResponse = {}", getResponse);
        }

        // Called when the whole GetRequest fails.
        @Override
        public void onFailure(Exception e) {
            logger.error("", e);
        }
    };

    public static void printIndexResponse(IndexResponse indexResponse) {
        logger.info("indexResponse = {}", indexResponse);
        String index = indexResponse.getIndex();
        String type = indexResponse.getType();
        String id = indexResponse.getId();
        long version = indexResponse.getVersion();
        logger.info("表:{}，类型：{}，ID:{},Version:{}", index, type, id, version);
        if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
            // Handle (if needed) the case where the document was created for the first time
            logger.info("新增");
        } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
            // Handle (if needed) the case where the document was rewritten as it was
            // already existing
            logger.info("更新");
        }
        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
        logger.info("分片总数：{}，成功数：{}", shardInfo.getTotal(), shardInfo.getSuccessful());
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
            // Handle the situation where number of successful shards is less than total
            // shards
            logger.warn("number of successful shards is less than total shards");
        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                // Handle the potential failures
                logger.warn("{}", failure);
                String reason = failure.reason();
                logger.warn("副本失败原因：" + reason);
            }
        }
    }

    public static Map<String, Object> printGetResponse(GetResponse getResponse) {
        logger.info("getResponse = {}", getResponse);
        String index = getResponse.getIndex();
        String type = getResponse.getType();
        String id = getResponse.getId();
        long version = getResponse.getVersion();
        logger.info("表:{}，类型：{}，ID:{},Version:{}", index, type, id, version);
        if (getResponse.isExists()) {
            String sourceAsString = getResponse.getSourceAsString();
            logger.info("getSourceAsString = {}", sourceAsString);
            Map<String, Object> sourceAsMap = getResponse.getSourceAsMap();
            logger.info("getSourceAsMap = {}", sourceAsMap);
            byte[] sourceAsBytes = getResponse.getSourceAsBytes();
            logger.info("getSourceAsBytes = {}", sourceAsBytes);
            return sourceAsMap;
        } else {
            /*
             * Note that although the returned response has 404 status code, a valid
             * GetResponse is returned rather than an exception thrown. Such response does
             * not hold any source document and its isExists method returns false.
             */
            logger.info("document {} don't exists", id);
            return null;
        }
    }

    public static void printSearchResponse(SearchResponse response) {
        logger.info("SearchResponse = {}", response);
        RestStatus status = response.status();
        logger.info("RestStatus = {}", status);
        TimeValue took = response.getTook();
        logger.info("take time:{}", took);
        Boolean terminatedEarly = response.isTerminatedEarly();
        boolean timedOut = response.isTimedOut();
        logger.info("timeout:{},isTerminatedEarly:{}", timedOut, terminatedEarly);
        int totalShards = response.getTotalShards();
        int successfulShards = response.getSuccessfulShards();
        int failedShards = response.getFailedShards();
        logger.info("shard,total:{},success:{},failed:{}", totalShards, successfulShards, failedShards);
        for (ShardSearchFailure failure : response.getShardFailures()) {
            // failures should be handled here
            logger.info("ShardSearchFailure = {}", failure);
        }
        SearchHits hits = response.getHits();
        TotalHits totalHits = hits.getTotalHits();
        float maxScore = hits.getMaxScore();
        logger.info("hit,total:{},maxScore:{}", totalHits, maxScore);
        // Retrieving SearchHits
        SearchHit[] searchHits = hits.getHits();
        for (SearchHit hit : searchHits) {
            // do something with the SearchHit
            String index = hit.getIndex();
            String type = hit.getType();
            String id = hit.getId();
            long version = hit.getVersion();
            float score = hit.getScore();
            logger.info("表:{}，类型：{}，ID:{},Version:{},score:{}", index, type, id, version, score);
            String sourceAsString = hit.getSourceAsString();
            System.err.println(sourceAsString);
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            logger.info("getSourceAsMap = {}", sourceAsMap);
            // Retrieving Highlighting
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();
            logger.info("HighlightFields = {}", highlightFields);
            for (Entry<String, HighlightField> entry : highlightFields.entrySet()) {
                String key = entry.getKey();
                HighlightField highlightField = entry.getValue();
                logger.info("key:{},value:{}", key, highlightField);
                String name = highlightField.getName();
                Text[] fragments = highlightField.fragments();
                logger.info("name:{},fragments:{}", name, fragments);
            }

            // Retrieving Aggregations
            Aggregations aggregations = response.getAggregations();
            if (aggregations != null) {
                for (Aggregation aggregation : aggregations) {
                    logger.info("aggregation = {}", aggregation);
                }
                Terms byCompanyAggregation = aggregations.get("by_company");
                Bucket elasticBucket = byCompanyAggregation.getBucketByKey("Elastic");
//                Avg averageAge = elasticBucket.getAggregations().get("average_age");
                Range range = aggregations.get("by_company");

                Map<String, Aggregation> aggregationMap = aggregations.getAsMap();
                Terms companyAggregation = (Terms) aggregationMap.get("by_company");

                List<Aggregation> aggregationList = aggregations.asList();

                for (Aggregation agg : aggregations) {
                    if (agg.getType().equals(TermsAggregationBuilder.NAME)) {
                        Bucket elasticBucketa = ((Terms) agg).getBucketByKey("Elastic");
                        long numberOfDocs = elasticBucketa.getDocCount();
                    }
                }
            }
            // Retrieving Suggest
            Suggest suggest = response.getSuggest();
            if (suggest != null) {
                TermSuggestion termSuggestion = suggest.getSuggestion("suggest_user");
                for (TermSuggestion.Entry entry : termSuggestion.getEntries()) {
                    for (TermSuggestion.Entry.Option option : entry) {
                        String suggestText = option.getText().string();
                    }
                }
            }

            // Retrieving profilingResults
            Map<String, ProfileShardResult> profilingResults = response.getProfileResults();
            if (profilingResults != null) {
                for (Map.Entry<String, ProfileShardResult> profilingResult : profilingResults.entrySet()) {
                    String key = profilingResult.getKey();
                    ProfileShardResult profileShardResult = profilingResult.getValue();
                    List<QueryProfileShardResult> queryProfileShardResults = profileShardResult
                            .getQueryProfileResults();
                    for (QueryProfileShardResult queryProfileResult : queryProfileShardResults) {
                        for (ProfileResult profileResult : queryProfileResult.getQueryResults()) {
                            String queryName = profileResult.getQueryName();
                            long queryTimeInMillis = profileResult.getTime();
                            List<ProfileResult> profiledChildren = profileResult.getProfiledChildren();
                        }
                        CollectorResult collectorResult = queryProfileResult.getCollectorResult();
                        String collectorName = collectorResult.getName();
                        Long collectorTimeInMillis = collectorResult.getTime();
                        List<CollectorResult> profiledChildren = collectorResult.getProfiledChildren();
                    }

                }
            }

        }
    }

    /**
     * 可选参数
     * 
     * @param request
     * @param routing
     * @param parent
     * @param seconds
     * @param refreshPolicy
     * @param version
     * @param versionType
     * @param opType
     * @param pipeline
     */
    public static void optionalArguments(IndexRequest request, String routing, Long seconds,
            WriteRequest.RefreshPolicy refreshPolicy, Long version, VersionType versionType,
            DocWriteRequest.OpType opType, String pipeline) {
        if (routing != null) {
            request.routing(routing);
        }
        /** Timeout to wait for primary shard to become available */
        if (seconds != null) {
            request.timeout(TimeValue.timeValueSeconds(seconds));
            // request.timeout("1s");
        }
        if (refreshPolicy != null) {
            request.setRefreshPolicy(refreshPolicy);
        }
        // request.setRefreshPolicy("wait_for");
        if (version != null) {
            request.version(version);
        }
        if (versionType != null) {
            request.versionType(versionType);
        }
        if (opType != null) {
            request.opType(opType);
        }
        if (pipeline != null) {
            request.setPipeline(pipeline);
        }
    }

    /**
     * 可选参数
     * 
     * @param request
     * @param routing
     * @param parent
     * @param version
     * @param versionType
     * @param preference
     * @param realtime
     * @param refresh
     * @param fetchSource
     * @param includes
     * @param excludes
     */
    public static void optionalArguments(GetRequest request, String routing, Long version, VersionType versionType,
            String preference, Boolean realtime, Boolean refresh, boolean fetchSource, String[] includes,
            String[] excludes) {
        if (fetchSource) {
            request.fetchSourceContext(new FetchSourceContext(true, includes, excludes));
        } else {
            // Disable source retrieval, enabled by default
            request.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE);
        }
        if (routing != null) {
            request.routing(routing);
        }
        if (preference != null) {
            request.preference(preference);
        }
        if (realtime != null) {
            // Set realtime flag to false (true by default)
            request.realtime(realtime);
        }
        if (refresh != null) {
            // Perform a refresh before retrieving the document (false by default)
            request.refresh(true);
        }
        if (version != null) {
            request.version(version);
        }
        if (versionType != null) {
            request.versionType(versionType);
        }
    }
}
